package net.xuele.learn.flink.keyedState;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 *
 * ETL（抽取、转换、加载）管道任务。从一个或多个数据源获取数据，进行一些转换操作和信息补充，将结果存储起来
 *
 * 实现一个要去重的事件数据流，对每个键只保留第一个事件
 */
public class KeyedState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new EventSource())
                .keyBy(Event::getKey)
                .flatMap(new Deduplicator())
                .print();
        env.execute();

    }

    static class EventSource implements SourceFunction<Event> {

        private static final long serialVersionUID = 6025713198203531909L;

        /**
         * 生成Event，伪代码实现
         *
         * @param sourceContext
         * @throws Exception
         */
        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            int i = 0;
            while (true) {
                Event event = new Event();
                event.setKey("key " + i);
                event.setTimestamp(System.currentTimeMillis());
                i++;
                sourceContext.collect(event);
            }
        }

        @Override
        public void cancel() {
            // do nothing
        }
        /**
         * 
         */
    }
}
